﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace HYService.RabbitMQ
{
    public partial class RabbitHub
    {
        /// <summary>
        /// 订阅队列处理消息
        /// </summary>
        /// <typeparam name="T">消息类型</typeparam>
        /// <param name="queue">队列名称</param>
        /// <param name="onMessage">处理消息的委托</param>
        /// <param name="noAck">是否禁用消费者的确认机制</param>
        /// <param name="onConfigure">通道的配置委托</param>
        public static EventingBasicConsumer Subscribe<T>(string queue, Action<T, EventingBasicConsumer, BasicDeliverEventArgs> onMessage, bool noAck = false, Action<IModel> onConfigure = null) where T : class
        {
            try
            {
                var connection = RabbitBus.CreateConnection();
                var channel = connection.CreateModel();
                channel.BasicQos(0, 1, false);
                channel.QueueDeclare(queue, true, false, false, null);
                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (sender, e) =>
                {
                    try
                    {
                        var json = Encoding.UTF8.GetString(e.Body);
                        var message = JsonConvert.DeserializeObject<T>(json);
                        onMessage(message, sender as EventingBasicConsumer, e);
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine("subscribe error occured:" + ex.Message);
                        channel.Requeue(e);
                    }
                    finally
                    {
                        if (!noAck)
                        {
                            channel.Ack(e);
                        }
                    }
                };

                if (onConfigure != null)
                {
                    onConfigure(channel);
                }

                channel.BasicConsume(queue, noAck, consumer);

                return consumer;
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }

        /// <summary>
        /// 订阅队列处理消息
        /// </summary>
        /// <typeparam name="T">消息类型</typeparam>
        /// <param name="queue">队列名称</param>
        /// <param name="onMessage">处理消息的委托</param>
        /// <param name="noAck">是否禁用消费者的确认机制</param>
        /// <param name="onConfigure">通道的配置委托</param>
        public static EventingBasicConsumer Subscribe<T>(string queue, Action<T> onMessage, bool noAck = false, Action<IModel> onConfigure = null) where T : class
        {
            try
            {
                var connection = RabbitBus.CreateConnection();
                var channel = connection.CreateModel();
                channel.BasicQos(0, 1, false);
                channel.QueueDeclare(queue, true, false, false, null);
                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (sender, e) =>
                {
                    try
                    {
                        var json = Encoding.UTF8.GetString(e.Body);
                        var message = JsonConvert.DeserializeObject<T>(json);
                        onMessage(message);
                    }
                    catch (Exception ex)
                    {
                        Console.WriteLine("subscribe error occured:" + ex.Message);
                        channel.Requeue(e);
                    }
                    finally
                    {
                        if (!noAck)
                        {
                            channel.Ack(e);
                        }
                    }
                };

                if (onConfigure != null)
                {
                    onConfigure(channel);
                }

                channel.BasicConsume(queue, noAck, consumer);

                return consumer;
            }
            catch (Exception ex)
            {
                throw ex;
            }
        }
    }
}
